package zk_01.zookeeper.sample;

import java.io.IOException;
import java.util.List;
import java.util.Random;

import org.apache.log4j.Logger;
import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/*
 * 异步获得管理权
 */
public class MasterUnSyn implements Watcher{
	
	static Logger logger = Logger.getLogger(MasterUnSyn.class);
	
	private String hostPort;
	private String leader;
	private boolean isLeader;
	private ZooKeeper zk;
	private String serverId;
	
	
	public static void main(String[] args){
		MasterUnSyn master = new MasterUnSyn("192.168.1.113:2181");
		//MasterUnSyn master = new MasterUnSyn("10.40.12.196:2181");
		try {
			master.startZK();
			master.runForMaster();
			master.bootstrap();
			master.getWorkers();
			master.getTask();
			Thread.sleep(1500000);
			master.stopZK();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	
	
	public MasterUnSyn() {}
	
	public MasterUnSyn(String hostPort){
		this.hostPort=hostPort;
	}
	//建立会话
	public void startZK() throws IOException{
		zk = new ZooKeeper(hostPort,15000,this);
	}
	//关闭会话
	public void stopZK() throws InterruptedException{
		zk.close();
	}
	
	StringCallback masterCreateCallBack = new StringCallback(){

		/*
		 * rc:执行命令请求返回的结果，将其转换为CODE枚举类型的，如果值为0表示成功，否则是KeeperException的编码值
		 * path:传给create命令的path参数值
		 * ctx:传给命令的上下文参数，如data
		 * name:创建的znode节点名称
		 */
		public void processResult(int rc, String path, Object ctx, String name) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				checkMaster();
				return;
			case OK:
				isLeader = true;
				masterExists();
				break;
			case NODEEXISTS:
				System.out.println("Leader was exists");
				masterExists();
			default:
				isLeader = false;
			}
			System.out.println("I'm "+(isLeader?"":"not ")+"the Leader");
		}
	};
	

	StringCallback createParentCallback = new StringCallback(){

		public void processResult(int rc, String path, Object ctx, String name) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				createParent(path,(byte[]) ctx);
				break;
			case OK:
				System.out.println("parent created");
			case NODEEXISTS:
				System.out.println("parent already registerd:"+path);
				break;
			default:
				System.out.println("something was wrong"+KeeperException.create(Code.get(rc),path));
			}
		}
	};
	
	DataCallback masterCheckCallback = new DataCallback(){

		public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				checkMaster();
				return;
			case NONODE:
				runForMaster();
				return;
			}
		}
	};
	
	Watcher masterExistsWatcher = new Watcher(){

		public void process(WatchedEvent event) {
			if(event.getType().equals(EventType.NodeDeleted)){
				System.out.println("The old leader is timeout");
				runForMaster();
			}
		}
		
	};
	
	/*
	 * 注册主节点发生变化监听的回调对象
	 * 回调对象逻辑：如果发生网络连接丢失，则重新注册；如果返回成功，判断ZNODE是否存在，存在注册成功，否则重新竞选主节点；其他异常根据获取的数据竞争主节点
	 * 
	 * */
	StatCallback masterExistsCallback = new StatCallback(){

		public void processResult(int rc, String path, Object ctx, Stat stat) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				masterExists();
				break;
			case OK:
				if(stat == null){
					runForMaster();
				}
				break;
			default:
				checkMaster();
				break;
			}
		}
		
	};
	
	/*
	 * 对主节点注册监听
	 */
	public void masterExists(){
		zk.exists("/master", masterExistsWatcher, masterExistsCallback, null);
	}

	/*
	 * 竞选主节点
	 */
	public void runForMaster(){
		Random random = new Random();
		serverId = Integer.toString(random.nextInt());
		zk.create("/master", serverId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL, masterCreateCallBack, null);
	}

	/*
	 * 检查ZK上是否已经有选举出来的主节点，如果没有就竞选主节点
	 */
	void checkMaster(){
		zk.getData("/master", false,masterCheckCallback,null);
	}
	
	/*
	 * 用于创建主从模式中的其他根节点的公用方法
	 */
	public void createParent(String path,byte[] data){
		zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, createParentCallback, data);
	}
	
	/*
	 * 创建主从模式中的其他根节点
	 */
	public void bootstrap(){
		createParent("/workers",new byte[0]);
		createParent("/assign",new byte[0]);
		createParent("/tasks",new byte[0]);
		createParent("/status",new byte[0]);
	}

	public void process(WatchedEvent event) {
		System.out.println(event);
	}
		
	
	//#####################################################观察从节点变化###############################################################
	/*
	 * 如果有节点删除或者加入，需要更新本地缓存
	 * ——如果是删除操作，还需要处理已经分配给从节点的任务
	 * */
	WorkerCache cache = new WorkerCache();
	
	Watcher getWorkdersWatcher = new Watcher(){

		public void process(WatchedEvent event) {
			logger.info("getWorkdersWatcher events:"+event);
			logger.info("==:"+(event.getType() == EventType.NodeChildrenChanged));
			logger.info("equals:"+(event.getType().equals(EventType.NodeChildrenChanged)));
			//收到通知，判断事件类型是不是子节点变化，是就刷新本地缓存
			if(event.getType() == EventType.NodeChildrenChanged){
				getWorkers();
			}
		}
	};
	
	ChildrenCallback getWorkersCallback = new ChildrenCallback(){

		public void processResult(int rc, String path, Object ctx,
				List<String> children) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				getWorkers();
				break;
			case OK:
				List<String> dList = cache.update(children);
				logger.info("this "+dList+" znode was deleted");
				reSendTask(dList);
				break;
			default:
				logger.info("get children failed");	
			}
		}		
	};
	
	private void reSendTask(List<String> dList) {
		//循环已失去通信的从节点，将分配给他们的任务重新分配
		for(String sName:dList){
			getTaskByName(sName);
		}
	}
	
	public void getWorkers(){
		logger.info("getWorkers start");
		zk.getChildren("/workers", getWorkdersWatcher, getWorkersCallback, null);
	}
	
	ChildrenCallback getDistributedTaskCallback = new ChildrenCallback(){

		public void processResult(int rc, String path, Object ctx,
				List<String> children) {
			logger.info("getDistributedTaskCallback path:"+path);
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				getTaskByName((String)ctx);
				break;
			case OK:
				distributeAssignedTask(children,(String)ctx);
				break;
			default:
				logger.info("getDistributedTaskCallback failed , "+Code.get(rc));
			}
		}
		
	};
	
	private void getTaskByName(String sName){
		List<String> lTask = null;
		zk.getChildren("/assign/"+sName, false, getDistributedTaskCallback, sName);
	}
	
	//#####################################################观察任务变化###############################################################
	/*
	 * 加入新任务，主节点从客户端列表中随机找一个并分派任务给他
	 * 1.删除任务
	 * 2.把任务与客户端绑定
	 * */
	

	
	public void getTask(){
		logger.info("-------getTask");
		zk.getChildren("/tasks", getTaskWatcher, getTaskCallback, null);
	}
	
	Watcher getTaskWatcher = new Watcher(){

		public void process(WatchedEvent event) {
			logger.info("getTaskWatcher event:"+event);
			if(event.getType() == EventType.NodeChildrenChanged){
				getTask();
			}
		}
	};
	ChildrenCallback getTaskCallback = new ChildrenCallback(){

		public void processResult(int rc, String path, Object ctx,
				List<String> children) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				getTask();
				break;
			case OK:
				distributeTask(children);
				break;
			default:
				logger.info("getTask failed");
				break;
			}
		}
		
	};
	
	DataCallback getAssignedTaskDataCallback = new DataCallback(){

		public void processResult(int rc, String path, Object ctx, byte[] data,
				Stat stat) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				getAssignedTaskData(path,(String)ctx);
				break;
			case OK:
				removeAndAsignTask(path,new String(data),(String)ctx,stat);
				break;
			default:
				logger.info("getAssignedTaskDataCallback failed");
				break;
			}
		}
		
	};
	
	public void distributeAssignedTask(List<String> children,String workerPath){
		//分配任务给从节点
		for(String tName:children){
			getAssignedTaskData(workerPath+"/"+tName,tName);
		}
		
		//待已失效的从节点对应任务重新分配完毕，从/assign/下删除对应从节点
		//##########################################待补充###################################################
	}
	
	public void getAssignedTaskData(String tPath,String tName){
		zk.getData("/assign/"+tPath, false, getAssignedTaskDataCallback, tName);
	}
	
	public void removeAndAsignTask(String path,String tData,String tName,Stat stat){
		deleteAssignedTask(path,stat.getVersion());
		String workerName = cache.getWorkerName();
		assignTask(workerName+"/"+tName,tData);
	}
	
	VoidCallback deleteAssignedTaskCallback = new VoidCallback(){

		public void processResult(int rc, String path, Object ctx) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				deleteAssignedTask(path,((Integer)ctx).intValue());
				break;
			case OK:
				logger.info("delete /assign/"+path+" successed");
				break;
			default:
				logger.info("delete /assign/"+path+" unSuccessed");
				break;
			}
		}
	};
	
	public void deleteAssignedTask(String oldPath,int version){
		zk.delete(oldPath, version, deleteAssignedTaskCallback, new Integer(version));
	}
	
	StringCallback assignTaskCallback = new StringCallback(){

		public void processResult(int rc, String path, Object ctx, String name) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				assignTask(path,(String)ctx);
				break;
			case OK:
				logger.info("assign /assign/"+path+" successed");
				break;
			default:
				logger.info("assign /assign/"+path+" unSuccessed");
				break;
			}
		}
		
	};
	
	public void assignTask(String tPath,String tData){
		logger.info("assignTask /assign/"+tPath+", data:"+tData);
		zk.create("/assign/"+tPath, tData.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, assignTaskCallback, tData);
	}
	
	public void distributeTask(List<String> children){
		for(String sTask:children){
			deleteAndAssignNewTask(sTask);
		}
	}
	
	public void deleteAndAssignNewTask(String sTask){
		getTaskVersion(sTask);
	}
	
	DataCallback getTaskVersionCallback = new DataCallback(){

		public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				getTaskVersion((String)ctx);
				break;
			case OK:
				logger.info("deleteTask "+path);
				deleteTask(path,stat.getVersion());
				for(;;){
					String workerName = cache.getWorkerName();
					logger.info("getTaskVersionCallback workerName "+workerName);
					if(workerName!=null){
						assignTask(workerName+"/"+path.substring(path.lastIndexOf("/")+1,path.length()),new String(data));
						break;
					}
				}
				break;
			default:
				logger.info("getTaskVersionCallback failed");
				break;
			}
		}
	};
	
	public void getTaskVersion(String sTask){
		logger.info("getTaskVersion start "+sTask);
		zk.getData("/tasks/"+sTask, false, getTaskVersionCallback, sTask);
	}
	
	VoidCallback deleteTaskCallback = new VoidCallback(){

		public void processResult(int rc, String path, Object ctx) {
			switch(Code.get(rc)){
			case CONNECTIONLOSS:
				deleteTask(path,((Integer)ctx).intValue());
				break;
			case OK:
				logger.info("deleteTaskCallback successed");
				break;
			default:
				logger.info("deleteTaskCallback unSuccessed");
				break;
			}
		}
		
	};
	
	public void deleteTask(String sTask,int version){
		zk.delete(sTask, version, deleteTaskCallback, new Integer(version));
	}

}
